#pragma once
#include <muduo/net/TcpClient.h>
#include <muduo/net/TcpServer.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/Buffer.h>
#include <muduo/net/EventLoopThread.h>
#include <muduo/base/CountDownLatch.h>
#include <functional>
#include <mutex>
#include <unordered_map>

#include "abstract.hpp"
#include "fields.hpp"
#include "detail.hpp"
#include "message.hpp"
namespace bitrpc
{
  class MuduoBuffer : public BaseBuffer
  {
  public:
    using ptr = std::shared_ptr<MuduoBuffer>;

    MuduoBuffer(muduo::net::Buffer *buf) : _buf(buf)
    {
    }
    virtual ssize_t readableSize() override // 返回缓冲区中有多少字节数据
    {
      return _buf->readableBytes();
    }
    virtual int32_t peekInt32() override // 读取缓冲区前四字节数据
    {
      return _buf->peekInt32(); // muduobuffer读取缓冲区的前四字节时会进行网络字节序到主机字节序的转换
    }
    virtual void retrieveInt32() override // 移除四个字节
    {
      _buf->retrieveInt32();
    }
    virtual int32_t readInt32() override // 读取缓冲区前四字节数据并移除四个字节
    {
      return _buf->readInt32();
    }
    virtual std::string retrieveAsString(size_t len) override // 将缓冲区数据全部读取出来
    {
      return _buf->retrieveAsString(len);
    }

  private:
    muduo::net::Buffer *_buf;
  };
  class BufferFactory
  {
  public:
    template <typename... Args>
    static BaseBuffer::ptr create(Args &&...args)
    {
      return std::make_shared<MuduoBuffer>(std::forward<Args>(args)...);
    }
  };
  class MuduoProtocol : public BaseProtocol
  {
  public:
    // 格式：Len--Value
    // Len--mtype--idlen--id--body
    using ptr = std::shared_ptr<MuduoProtocol>;
    virtual bool canProcessed(const BaseBuffer::ptr &buf) override
    {
     
       // 判断缓冲区中的数据量是否⾜够⼀条消息的处理---报文完整性检查
      if (buf->readableSize() < lenFieldsLength)
      {
        DLOG("当前缓冲区内的数据不够四字节");
        return false;
      }
      int32_t total_len = buf->peekInt32();

    //    ssize_t sz = buf->readableSize();
    //  std:: string str = buf->retrieveAsString(sz);
    //  std::cout<<str<<std::endl;
    //  DLOG("total_len + lenFieldsLength:%d",total_len + lenFieldsLength);
    //  DLOG("buf->readableSize():%d",buf->readableSize());
      if((total_len + lenFieldsLength > buf->readableSize()))
      {
        DLOG("数据量不足");
        return false;
      }
      return true;
    }
    virtual bool onMessage(const BaseBuffer::ptr &buf, BaseMessage::ptr &msg) override // 处理缓冲区内数据(反序列化)
    {
      // Len--mtype--idlen--id--body
      // 当调⽤onMessage的时候，默认认为缓冲区中的数据⾜够⼀条完整的消息
      int32_t total_len = buf->readInt32();              // 读取数据总长度
      MessageType mtype = (MessageType)buf->readInt32(); // 读取消息类型
      int32_t idlen = buf->readInt32();                  // 读取id长度
      std::string id = buf->retrieveAsString(idlen);
      int32_t body_len = total_len - mtypeFieldsLength - idlen - idlenFieldsLength;
      string body = buf->retrieveAsString(body_len);

      msg = MessageFactory::create(mtype);
      if (msg.get() == nullptr)
      {
        ELOG("消息类型错误，构造消息对象失败！")
        return false;
      }
      if (msg->DeSerialize(body) == false)
      {
        ELOG("消息正⽂反序列化失败！");
        return false;
      }
      msg->SetId(id);
      msg->SetMessageType(mtype);
      return true;
    }
    virtual std::string serialize(const BaseMessage::ptr &msg) override // 对处理后的消息进行序列化
    {
      // Len--mtype--idlen--id--body
      string body = msg->Serialize();
      int32_t mtype = htonl((int32_t)msg->messageType());
      string id = msg->rid();
      int32_t idlen = htonl(id.size());
      int32_t h_total_len = mtypeFieldsLength + id.size() + idlenFieldsLength + body.size();
      int32_t n_total_len = htonl(h_total_len);

      string result;
      
      result.append((char *)&n_total_len, lenFieldsLength);
      result.append((char *)&mtype, mtypeFieldsLength);
      result.append((char *)&idlen, idlenFieldsLength);
      result.append(id);
      result.append(body);
      return result;
    }

  private:
    const int lenFieldsLength = 4;
    const int mtypeFieldsLength = 4;
    const int idlenFieldsLength = 4;
  };
  class ProtocolFactory
  {
  public:
    template <typename... Args>
    static BaseProtocol::ptr create(Args &&...args)
    {
      return std::make_shared<MuduoProtocol>(std::forward<Args>(args)...);
    }
  };
  class MuduoConnection : public BaseConnection
  {
  public:
    using ptr = std::shared_ptr<MuduoConnection>;
    MuduoConnection(const muduo::net::TcpConnectionPtr &conn, const BaseProtocol::ptr &pro)
        : _pro(pro),
          _conn(conn)
    {
    }
    virtual void send(const BaseMessage::ptr &msg) override
    {
      string message = _pro->serialize(msg);
      _conn->send(message);
    }
    virtual void shutdown() override
    {
      _conn->shutdown();
    }
    virtual bool connected() override
    {
      return _conn->connected();
    }

  private:
    BaseProtocol::ptr _pro;
    muduo::net::TcpConnectionPtr _conn;
  };
  class ConnectionFactory
  {
  public:
    template <typename... Args>
    static BaseConnection::ptr create(Args &&...args)
    {
      return std::make_shared<MuduoConnection>(std::forward<Args>(args)...);
    }
  };
  class MuduoServer : public BaseServer
  {
  public:
    using ptr = std::shared_ptr<MuduoServer>;
    MuduoServer(int port)
        // 给TcpServer初始化，首先要有一个EventLoop，然后是InetAddress，端口和IP地址，最后是选项
        : _server(&_basicloop, muduo::net::InetAddress("0.0.0.0", port),
                  "DictServer", muduo::net::TcpServer::Option::kReusePort),
          _pro(ProtocolFactory::create())
    {
    }
    virtual void start() override
    {
      // 注册回调函数，连接建立和断开连接，以及收到消息时的回调函数
      _server.setConnectionCallback(std::bind(&MuduoServer::OnConnection, this, std::placeholders::_1));
      // _server.setMessageCallback(std::bind(&DictServer::OnMessage,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
      _server.setMessageCallback(std::bind(&MuduoServer::OnMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
      _server.start();   // 先启动服务器，注册监听事件，创建线程池
      _basicloop.loop(); // 再开始死循环事件监控
    }

  private:
    void OnConnection(const muduo::net::TcpConnectionPtr &conn)
    {
      if (conn->connected())
      {
        std::cout << "建立连接" << std::endl;
        auto muduo_conn = ConnectionFactory::create(conn, _pro);
        {
          // 在多线程环境下，多个线程可能同时访问和修改 _conns 这个集合，
          // 如果不加锁，可能会导致数据竞争，比如一个线程正在向集合中插入数据，另一个线程也在操作同一个集合，
          // 这样就会导致数据不一致或者程序崩溃
          // std::unique_lock<std::mutex> 是一个智能锁类，它会在构造时自动获取锁，在析构时自动释放锁
          unique_lock<std::mutex> _lock(mutex);
          _conns.insert(std::make_pair(conn, muduo_conn));
        }
        if (_cb_connection)
          _cb_connection(muduo_conn);
      }
      else
      {
        std::cout << "断开连接" << std::endl;
        BaseConnection::ptr muduo_conn;
        {
          std::unique_lock<std::mutex> lock(_mutex);
          if (_conns.count(conn) == 0)
            return;
          muduo_conn = _conns[conn];
          _conns.erase(conn);
        }
        if (_cb_close)
          _cb_close(muduo_conn);
      }
    }

    void OnMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp time)
    {
      DLOG("连接有数据到来，开始处理！");
      BaseBuffer::ptr case_buf = BufferFactory::create(buf);
      while (true)
      {
        if (_pro->canProcessed(case_buf) == false)
        {
          // 数据不足
          // 有一种特殊情况是：数据量很大，超出了缓冲区容量，但是又不足一条消息的

          if (case_buf->readableSize() > maxDataSize)
          {
            ELOG("缓冲区中数据过⼤！");
            conn->shutdown();
            return;
          }
          DLOG("数据量不⾜！")
          break;
        }
        DLOG("缓冲区中数据可处理！")
        BaseMessage::ptr msg;
        bool ret = _pro->onMessage(case_buf, msg);
        if (!ret)
        {
          conn->shutdown();
          ELOG("缓冲区中数据错误！");
          return;
        }
        DLOG("消息反序列化成功！")
        BaseConnection::ptr base_conn;
        {
          std::unique_lock<std::mutex> lock(_mutex);
          if (_conns.count(conn) == 0)
          {
            conn->shutdown();
            return;
          }
          base_conn = _conns[conn];
        }
        DLOG("调⽤回调函数进⾏消息处理！")
        if (_cb_message)
          _cb_message(base_conn, msg);
      }
    }

  private:
    const size_t maxDataSize = (1 << 16);
    BaseProtocol::ptr _pro;
    muduo::net::TcpServer _server;
    muduo::net::EventLoop _basicloop;
    std::mutex _mutex;
    std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::ptr> _conns;
  };

  class ServerFactory
  {
  public:
    template <typename... Args>
    static BaseServer::ptr create(Args &&...args)
    {
      return std::make_shared<MuduoServer>(std::forward<Args>(args)...);
    }
  };
  class MuduoClient : public BaseClient
  {
  public:
    using ptr = std::shared_ptr<MuduoClient>;
    MuduoClient(const std::string &serverip, int serverport)
        : _pro(ProtocolFactory::create()),
          _client(_loop, muduo::net::InetAddress(serverip, serverport), "DictClient"),
          _loop(_threadloop.startLoop()),
          _latch(1)
    {
    }
    virtual void connect() override // 建立连接
    {
      _client.setConnectionCallback(std::bind(&MuduoClient::OnConnection, this, std::placeholders::_1));
      _client.setMessageCallback(std::bind(&MuduoClient::OnMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
      _client.connect(); // 建立连接，由于链接是异步且非阻塞，所以有可能出现连接还没建立但已经开始读写的情况，所以进程需要等待连接建立完毕
      _latch.wait();     // 只要计数器不为0，就阻塞
      // _loop.loop();//loop不能直接调用，因为会死环，无法调用send发送数据，所以要借助其他线程来调用loop
    }
    virtual void shutdown() override // 关闭连接
    {
      _client.disconnect();
    }
    virtual bool send(const BaseMessage::ptr &msg) override // 发送消息
    {
      if (connected() == false)
      {
        ELOG("连接断开");
        return false;
      }
      _conn->send(msg);
      return true;
    }
    virtual BaseConnection::ptr connection() override // 获取连接对象
    {
      return _conn;
    }
    virtual bool connected() override // 判断连接是否正常
    {
      return _conn && _conn->connected();
    }

  private:
    void OnConnection(const muduo::net::TcpConnectionPtr &conn)
    {
      if (conn->connected())
      {
        std::cout << "建立连接" << std::endl;
        _latch.countDown(); // 计数器减一，代表有一个连接建立完毕
        _conn = ConnectionFactory::create(conn, _pro);
      }
      else
      {
        std::cout << "断开连接" << std::endl;
        _conn.reset();
      }
    }
    void OnMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp time)
    {
      BaseBuffer::ptr case_buf = BufferFactory::create(buf);
      while (true)
      {
        if (_pro->canProcessed(case_buf) == false)
        {
          // 数据不足
          // 有一种特殊情况是：数据量很大，超出了缓冲区容量，但是又不足一条消息的

          if (case_buf->readableSize() > maxDataSize)
          {
            ELOG("缓冲区中数据过⼤！");
            conn->shutdown();
            return;
          }
          break;
        }
        BaseMessage::ptr msg;
        bool ret = _pro->onMessage(case_buf, msg);
        if (!ret)
        {
          conn->shutdown();
          ELOG("缓冲区中数据错误！");
          return;
        }

        if (_cb_message)
          _cb_message(_conn, msg);
      }
    }

  protected:
    const size_t maxDataSize = (1 << 16);

    BaseProtocol::ptr _pro;
    muduo::net::EventLoopThread _threadloop; // 通过线程的EventLoop监听事件，一旦实例化，会自动创建线程，启动EventLoop监听
    muduo::net::EventLoop *_loop;
    std::mutex _mutex;

    muduo::net::TcpClient _client;
    BaseConnection::ptr _conn;
    muduo::CountDownLatch _latch; // 计数器，数目是客户端的链接个数
  };
  class ClientFactory
  {
  public:
    template <typename... Args>
    static BaseClient::ptr create(Args &&...args)
    {
      return std::make_shared<MuduoClient>(std::forward<Args>(args)...);
    }
  };
}